package cz.data.domain.metadata.service.impl;

import cz.data.common.database.CacheDataSourceBean;
import cz.data.common.support.db.DbProperty;
import cz.data.common.database.utils.DbQueryUtil;
import cz.data.common.exception.DataException;
import cz.data.common.utils.ThrowableUtil;
import cz.data.domain.metadata.concurrent.CallableTemplate;
import cz.data.domain.metadata.concurrent.JdbcSelectTemplate;
import cz.data.domain.metadata.model.MetadataUtil;
import cz.data.domain.metadata.model.dto.SqlConsoleDto;
import cz.data.domain.metadata.model.entity.MetadataSourceEntity;
import cz.data.domain.metadata.model.vo.SqlConsoleVo;
import cz.data.domain.metadata.service.MetadataSourceService;
import cz.data.domain.metadata.service.SqlConsoleService;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.Statements;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class SqlConsoleServiceImpl implements SqlConsoleService {

    @Autowired
    private MetadataSourceService metadataSourceService;

    private static final Map<String, List<Connection>> connectionMap = new ConcurrentHashMap<>();

    @Override
    public List<SqlConsoleVo> sqlRun(SqlConsoleDto sqlConsoleDto) {
        sqlStop(sqlConsoleDto);
        String sqlKey = sqlConsoleDto.getSqlKey();
        Statements stmts;
        try {
            stmts = CCJSqlParserUtil.parseStatements(sqlConsoleDto.getSqlText());
        } catch (JSQLParserException e) {
            log.error("全局异常信息ex={}, StackTrace={}", e.getMessage(), ThrowableUtil.getStackTrace(e));
            throw new DataException("SQL语法有问题，解析出错");
        }
        List<Statement> sqls = stmts.getStatements();
        if (CollectionUtils.isEmpty(sqls)) {
            throw new DataException("未解析到SQL语句");
        } else if (CollectionUtils.size(sqls) > 5) {
            throw new DataException("SQL语句语句过多请稍后再试!");
        }
        MetadataSourceEntity dataSource = metadataSourceService.getMetadataSourceById(sqlConsoleDto.getSourceId());
        if (dataSource == null) {
            throw new DataException("SQL工作台查询数据源出错");
        }
        DbQueryUtil.requireSupportJdbc(dataSource.getDbType());
        DbProperty property = MetadataUtil.dbProperty(dataSource);
        DataSource source = CacheDataSourceBean.createDataSource(property);
        // 定义计数器
        final CountDownLatch latch = new CountDownLatch(sqls.size());
        // 定义固定长度的线程池
        ExecutorService executorService = null;
        // Future用于获取结果
        List<SqlConsoleVo> result = new ArrayList<>();
        try {
            executorService = Executors.newFixedThreadPool(sqls.size());
            // Callable用于产生结果
            List<CallableTemplate<SqlConsoleVo>> tasks = new ArrayList<>();
            List<Connection> conns = new ArrayList<>();
            for (Statement sql : sqls) {
                Connection conn = source.getConnection();
                conns.add(conn);
                JdbcSelectTemplate jdbcSelectTemplate = new JdbcSelectTemplate(latch, conn, sql.toString());
                tasks.add(jdbcSelectTemplate);
            }
            connectionMap.put(sqlKey, conns);
            List<Future<SqlConsoleVo>> futures = executorService.invokeAll(tasks);
            // 主线程阻塞，等待所有子线程执行完成
            latch.await();
            // 处理线程返回结果
            for (Future<SqlConsoleVo> future : futures) {
                result.add(future.get());
            }
        } catch (Exception e) {
            log.error("全局异常信息ex={}, StackTrace={}", e.getMessage(), ThrowableUtil.getStackTrace(e));
        } finally {
            if (Objects.nonNull(executorService)) {
                // 关闭线程池
                executorService.shutdown();
            }
            // 执行完清除
            connectionMap.remove(sqlKey);
        }
        return result;
    }

    @Override
    public void sqlStop(SqlConsoleDto sqlConsoleDto) {
        String sqlKey = sqlConsoleDto.getSqlKey();
        List<Connection> conns = connectionMap.get(sqlKey);
        if (CollectionUtils.isNotEmpty(conns)) {
            List<Exception> errors = new ArrayList<>();
            for (Connection conn : conns) {
                try {
                    if (null != conn && !conn.isClosed()) {
                        conn.close();
                    }
                } catch (SQLException e) {
                    errors.add(e);
                }
            }
            if (!errors.isEmpty()) {
                String errorMessage = errors.stream()
                        .map(Exception::getMessage)
                        .collect(Collectors.joining(";"));
                throw new DataException("SQL工作台停止出错:" + errorMessage);
            }
        }
    }
}
